package cn.remex.configuration.etcd;

import cn.remex.configuration.etcd.client.EtcdV3Client;
import cn.remex.configuration.etcd.schema.Event;
import cn.remex.core.exception.ServiceCode;
import cn.remex.core.util.Assert;
import cn.remex.core.util.Judgment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanWrapper;
import org.springframework.beans.BeanWrapperImpl;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Disc:
 * Created by xinxinran on 2017/7/24.
 */
public class EtcdConfigCenter {

    private static final Logger logger= LoggerFactory.getLogger(EtcdConfigCenter.class);

    private static final ConcurrentHashMap<String, EtcdV3Client> cachedClientMap = new ConcurrentHashMap<>(8);

    private static final ConcurrentHashMap<String, String> cachedConfigMap = new ConcurrentHashMap<>(512);

    private static final Map<String, Map<String, String>> cachedUsingConfigMap = new ConcurrentHashMap<>(8);

    private static final ConcurrentHashMap<String, List<Consumer<String>>> watchProcessorMap = new ConcurrentHashMap<>(512);

    private static final ConcurrentHashMap<Object, BeanWrapper> cachedBeanWarpper = new ConcurrentHashMap<>(512);

    private static final String ETCD_KEY_PREFIX = "etcd://";

    private static String watchKeyPrefix = null;

    private static Properties totalMergedProperties = null;

    private static Function<String, List<String>> getKeysFromPlaceHolder;

    @SuppressWarnings("unused")
    public static EtcdV3Client getClientInstance(){
        Assert.isTrue(cachedClientMap.containsKey("::default"), ServiceCode.ERROR, "没有可用的Etcd连接");
        return cachedClientMap.get("::default");
    }

    public static EtcdV3Client getClientInstance(String connString){
        connString = formatConnString(connString);

        if(cachedClientMap.containsKey(connString)){
            return cachedClientMap.get(connString);
        }

        EtcdV3Client client = new EtcdV3Client(connString);
        cachedClientMap.put(connString, client);

        if(!cachedClientMap.containsKey("::default")){
            cachedClientMap.put("::default", client);
        }
        return client;
    }

    public synchronized static void initEtcdConfig(String connString, String cachedPrefix, String watchPrefix, Function<String, List<String>> getKeysFunction) throws ExecutionException, InterruptedException {
        if(cachedClientMap.containsKey(connString)){
            return;
        }
        EtcdV3Client client = getClientInstance(connString);

        if(!Judgment.nullOrBlank(cachedPrefix)){
            cachedConfigMap.putAll(client.getRange(cachedPrefix).get());

            cachedConfigMap.forEach((key, value) -> {
                cachedUsingConfigMap.forEach((keyPrefix, map) -> {
                    if(key.startsWith(keyPrefix)){
                        map.put(key, value);
                    }
                });
            });
        }

        watchKeyPrefix = watchPrefix;
        getKeysFromPlaceHolder = getKeysFunction;

        if(!Judgment.nullOrBlank(watchKeyPrefix)){
            client.watchRange(watchKeyPrefix, changeInfo -> {
                List<Event> eventList = changeInfo.getEventsList();
                if (eventList != null) {
                    eventList.forEach(event -> {
                        String key = event.getKv().getKey().toStringUtf8();
                        String newVal = event.getKv().getValue().toStringUtf8();
                        cachedConfigMap.put(key, newVal);

                        cachedUsingConfigMap.forEach((keyPrefix, map) -> {
                            if(key.startsWith(keyPrefix)){
                                map.put(key, newVal);
                            }
                        });

                        if (watchProcessorMap.containsKey(key)) {

                            watchProcessorMap.get(key).forEach(processor -> processor.accept(newVal));
                        }
                    });
                }
            }, false);
        }
    }

    public static synchronized void setMergedProperties(Properties properties){
        totalMergedProperties = properties;
    }

    public static Map<String, String> getConfigMap(String keyPrefix){
        Assert.notNullAndEmpty(keyPrefix, ServiceCode.ERROR, "[EtcdConfigCenter::getConfigMap] keyPrefix can not be empty");
        if(cachedUsingConfigMap.containsKey(keyPrefix)){
            return cachedUsingConfigMap.get(keyPrefix);
        }

        Map<String, String> configMap = new HashMap<>();

        cachedConfigMap.forEach((key, value) -> {
            if (key.startsWith(keyPrefix)){
                configMap.put(key, value);
            }
        });

        cachedUsingConfigMap.put(keyPrefix, configMap);
        return configMap;
    }

    private static String formatConnString(String connString){
        String[] connInfo = connString.split(":");
        String host = connInfo[0];
        int port = 2379;
        int maxWatchCount = 20;
        if(connInfo.length > 1 && connInfo[1].matches("^\\d{2,6}$")){
            port = Integer.parseInt(connInfo[1]);
        }

        if(connInfo.length > 2 && connInfo[2].matches("^\\d{1,2}$")){
            maxWatchCount = Integer.parseInt(connInfo[2]);
        }

        return host + ":" + port + ":" + maxWatchCount;
    }

    public static String getConfigString(String connString, String key){
        if(cachedConfigMap.containsKey(key)){
            return cachedConfigMap.get(key);
        }

        try {
            return getClientInstance(connString).getValue(key).get();
        } catch (InterruptedException | ExecutionException e) {
            logger.warn("[EtcdConfigCenter::getConfigString] get etcd value error", e);
            return null;
        }
    }

    public static void registerWatching(String watchKey, Consumer<String> changeProcessor){
        if(!watchProcessorMap.containsKey(watchKey)){
            watchProcessorMap.put(watchKey, new ArrayList<>());
        }

        watchProcessorMap.get(watchKey).add(changeProcessor);
    }

    public static void setBeanProperty(Object bean, String propName, String value){
        if(!cachedBeanWarpper.containsKey(bean)){
            cachedBeanWarpper.put(bean, new BeanWrapperImpl(bean));
        }

        cachedBeanWarpper.get(bean).setPropertyValue(propName, value);
    }

    public static boolean isEtcdKey(String keyString){
        return !Judgment.nullOrBlank(keyString) && keyString.startsWith(ETCD_KEY_PREFIX);
    }

    public static String getEtcdKey(String keyString){
        if(isEtcdKey(keyString)){
            return keyString.substring(ETCD_KEY_PREFIX.length() - 1);
        }

        return keyString;
    }

    public static boolean canWatchEtcdKey(String keyString){
        return !Judgment.nullOrBlank(watchKeyPrefix)
                && !Judgment.nullOrBlank(keyString)
                && keyString.startsWith(ETCD_KEY_PREFIX + watchKeyPrefix.substring(1));
    }

    public static List<String> getEtcdKeysFromPlaceHolder(String placeholderString){
        if(getKeysFromPlaceHolder == null){
            return new ArrayList<>();
        }

        return getKeysFromPlaceHolder.apply(placeholderString).stream().map(key -> {
            String tmpVal = totalMergedProperties.getProperty(key);
            return Judgment.nullOrBlank(tmpVal) ? key : tmpVal;
        }).filter(EtcdConfigCenter::isEtcdKey).collect(Collectors.toList());
    }
}
